package com.exmaple.cdc.service;

import com.exmaple.cdc.listener.CDCMessageListener;
import com.exmaple.cdc.wrap.BinlogAddObj;
import com.exmaple.cdc.wrap.BinlogObj;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.ResultSetExtractor;

import java.io.Serializable;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.logging.Logger;

public class SysMQServiceImpl implements SysMQService {

    Logger logger = Logger.getLogger(SysMQServiceImpl.class.getName());

    private JdbcTemplate jdbcTemplate;

    public JdbcTemplate getJdbcTemplate() {
        return jdbcTemplate;
    }

    public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    /***
     * 把mq消息插入中间消息表 把 status --> 0
     * @param modelName
     * @param modelId
     * @param context
     * @return
     */
    @Override
    public int publish(String modelName, Serializable modelId, String table, String context) {
        String sql = "insert into " + table + " (model_id , model_name , context , create_time , update_time  , status) values (?, ? ,? , ? , ?, ?)";
        return jdbcTemplate.update(sql, modelId.toString(), modelName, context, new Date(), new Date(), 0);
    }


    /***
     * 消息的提交 把 status --> 1
     * @param binlogObj
     * @throws Exception
     */
    @Override
    public void commitObj(BinlogObj binlogObj) {
        if (binlogObj instanceof BinlogAddObj) {
            BinlogAddObj binlogAddObj = (BinlogAddObj) binlogObj;
            for (Map<String, Serializable> map : binlogAddObj.getRowsList()) {
                if (map.containsKey(CDCMessageListener.ID) && map.get(CDCMessageListener.ID) != null) {
                    Long id = (Long) map.get(CDCMessageListener.ID);
                    this.commit(binlogObj.getTableName(), id);
                }
            }
        }
    }

    /***
     * 消息的回滚，设置其消费失败和重试。把 status --> 2 , 让重试线程一直去消费
     * @param binlogObj
     */
    @Override
    public void rollBackObj(BinlogObj binlogObj) {
        if (binlogObj instanceof BinlogAddObj) {
            BinlogAddObj binlogAddObj = (BinlogAddObj) binlogObj;
            for (Map<String, Serializable> map : binlogAddObj.getRowsList()) {
                if (map.containsKey(CDCMessageListener.ID) && map.get(CDCMessageListener.ID) != null) {
                    Long id = (Long) map.get(CDCMessageListener.ID);
                    this.rollBack(binlogObj.getTableName(), id);
                }
            }
        }
    }

    /***
     * 消息的确认
     * @param id
     * @return
     */
    @Override
    public int commit(String table, Long id) {
        String sql = "update " + table + " set status = ? where id = ?";
        return jdbcTemplate.update(sql, 1, id);
    }

    /***
     * 消息的重试
     * @param table
     * @param id
     * @return
     */
    @Override
    public int rollBack(String table, Long id) {
        String sql = "update " + table + " set status = ? where id = ?";
        return jdbcTemplate.update(sql, 2, id);
    }

    /***
     * 设置消息的回滚
     * @param table
     * @param id
     * @param retryCount
     * @param retryTime
     * @return
     */
    @Override
    public int rollBack(String table, Long id, int retryCount, Date retryTime) {
        return 0;
    }

    /***
     * 查询未处理的消息
     * @param table
     * @param status
     * @return
     */
    @Override
    public List<Map<String, Object>> queryList(String table, String status) {
        String sql = "select * from " + table + " where status = " + status;
        return jdbcTemplate.query(sql, new ResultSetExtractor<List<Map<String, Object>>>() {
            @Override
            public List<Map<String, Object>> extractData(ResultSet rs) throws SQLException, DataAccessException {
                List<Map<String, Object>> list = new ArrayList<>();
                if (rs.next()) {
                    Map<String, Object> map = new HashMap<>();
                    map.put(CDCMessageListener.ID, rs.getString(CDCMessageListener.ID));
                    map.put(CDCMessageListener.MODEL_ID, rs.getString(CDCMessageListener.MODEL_ID));
                    map.put(CDCMessageListener.MODEL_NAME, rs.getString(CDCMessageListener.MODEL_NAME));
                    map.put(CDCMessageListener.CONTEXT, rs.getString(CDCMessageListener.CONTEXT));
                    map.put(CDCMessageListener.STATUS, rs.getString(CDCMessageListener.STATUS));
                    map.put(CDCMessageListener.RETRY_COUNT, rs.getString(CDCMessageListener.RETRY_COUNT));
                    map.put(CDCMessageListener.RETRY_TIME, rs.getString(CDCMessageListener.RETRY_TIME));
                    list.add(map);
                }
                return list;
            }
        });
    }

    /***
     * 查询数据
     * @param id
     * @return
     */
    @Override
    public Map<String, Object> queryById(String table, Long id) {
        String sql = "select * from " + table + " where id = " + id;
        return jdbcTemplate.query(sql, new ResultSetExtractor<Map<String, Object>>() {
            @Override
            public Map<String, Object> extractData(ResultSet rs) throws SQLException, DataAccessException {
                Map<String, Object> map = new HashMap<>();
                if (rs.next()) {
                    map.put(CDCMessageListener.ID, rs.getString(CDCMessageListener.ID));
                    map.put(CDCMessageListener.MODEL_ID, rs.getString(CDCMessageListener.MODEL_ID));
                    map.put(CDCMessageListener.MODEL_NAME, rs.getString(CDCMessageListener.MODEL_NAME));
                    map.put(CDCMessageListener.CONTEXT, rs.getString(CDCMessageListener.CONTEXT));
                    map.put(CDCMessageListener.STATUS, rs.getString(CDCMessageListener.STATUS));
                    map.put(CDCMessageListener.RETRY_COUNT, rs.getString(CDCMessageListener.RETRY_COUNT));
                    map.put(CDCMessageListener.RETRY_TIME, rs.getString(CDCMessageListener.RETRY_TIME));
                }
                return map;
            }
        });
    }


    /****
     * 判断消息是否消费完毕
     * @param binlogObj
     * @return
     */
    @Override
    public boolean isCommit(BinlogObj binlogObj) {
        boolean flag = true;
        for (Map<String, Serializable> map : ((BinlogAddObj) binlogObj).getRowsList()) {
            Long id = (Long) map.get(CDCMessageListener.ID);
            Map<String, Object> dataMap = this.queryById(binlogObj.getTableName(), id);
            if (dataMap.containsKey(CDCMessageListener.STATUS) && dataMap.get(CDCMessageListener.STATUS) != null) {
                Object status = dataMap.get(CDCMessageListener.STATUS);
                if (status == null) {
                    flag = false;
                    break;
                } else {
                    if (status.toString().equalsIgnoreCase(CDCMessageListener.INIT)) {
                        flag = false;
                        break;
                    }
                }
            }
        }
        return flag;
    }

    @Override
    public boolean isCommit(String table, Long id) {
        Map<String, Object> dataMap = this.queryById(table, id);
        Object status = dataMap.get(CDCMessageListener.STATUS);
        return status.toString().equalsIgnoreCase(CDCMessageListener.INIT);
    }
}
